package rabbitmq

import (
	"fmt"
	"log"

	"github.com/streadway/amqp"
)

//MQURL ..url:"amqp://账号:密码@地址:端口/虚拟主机"
const MQURL = "amqp://wd:123@192.168.0.127:5672/test"

//RabbitMQ ...
type RabbitMQ struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	//队列的名称
	QueueName string
	//交换器
	Exchange string
	//key
	Key string
	//连接信息
	Mqurl string
}

//NewRabbitMQ ..创建Rabbitmq实例
func NewRabbitMQ(queuename, exchange, key string) *RabbitMQ {
	rabbitmq := &RabbitMQ{
		QueueName: queuename,
		Exchange:  exchange,
		Key:       key,
		Mqurl:     MQURL,
	}
	var err error
	//创建rabbitmq连接
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.Err(err, "NewRabbitMQSimple-->创建连接错误")
	//创建隧道
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.Err(err, "NewRabbitMQSimple-->获取隧道错误")
	return rabbitmq
}

//Close ..断开连接
func (r *RabbitMQ) Close() {
	r.conn.Close()
	r.channel.Close()
}

//Err 错误日志
func (r *RabbitMQ) Err(err error, message string) {
	if err != nil {
		log.Fatal(message, err)
	}
}

//NewRabbitMQSimple ..简单模式实例
func NewRabbitMQSimple(queueName string) *RabbitMQ {
	return NewRabbitMQ(queueName, "", "")
}

//PublishSimple ..简单模式下的生产者
func (r *RabbitMQ) PublishSimple(message string) {
	//1.申请队列（不存在则创建），保证消息一定送达
	_, err := r.channel.QueueDeclare(
		r.QueueName,
		false, //控制消息是否持久化，重启服务器数据是否保留
		false, //是否自动删除，当最后一个消费者从队列离开，是否删除消息
		false, //排他性，false时紧自己可见，其他用户不能访问
		false, //是否阻塞，
		nil)   //额外的属性
	if err != nil {
		fmt.Println(err)
	}
	//2.发送消息到队列中
	r.channel.Publish(
		r.Exchange, //为空，默认交换机
		r.QueueName,
		false, //为true，根据exchange类型，和routkey规则，如果无法找到符合条件的队列，那么会将消息返还给发送者
		false, //为true，当队列没有绑定消费者，那么会将消息返还给发送者
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

//ConsumeSimple ..简单模式下，消费者
func (r *RabbitMQ) ConsumeSimple() {
	//申请队列
	_, err := r.channel.QueueDeclare(r.QueueName, false, false, false, false, nil)
	if err != nil {
		fmt.Println(err)
	}
	//接收消息
	mesgs, err := r.channel.Consume(
		r.QueueName,
		"",    //区分多个消费者
		true,  //是否自动应答，
		false, //排他性
		false, //为true，不能将同一个connection中发送的消息传递给这个connection中的消费者
		false, //是否设置为阻塞
		nil)
	if err != nil {
		fmt.Println(err)
	}
	forever := make(chan bool)
	go func() {
		for d := range mesgs {
			//实现我们要处理的逻辑函数
			log.Printf("Received a message :%s", d.Body)
		}
	}()
	log.Printf("[*] waiting for messages, to exit press ctrl+c!!!")
	<-forever
}

//NewRabbitMQPubSub ..订阅模式创建Rabbitmq实例
func NewRabbitMQPubSub(exchangeName string) *RabbitMQ {
	rbmq := NewRabbitMQ("", exchangeName, "")
	return rbmq
}

//PublishSub .. 订阅模式下生产者
func (r *RabbitMQ) PublishSub(message string) {
	//创建交换机(不存在则创建)
	err := r.channel.ExchangeDeclare(
		r.Exchange, //交换机名称
		"fanout",   //广播类型
		true,       //控制消息是否持久化，重启服务器数据是否保留
		false,      //是否自动删除，当最后一个消费者从队列离开，是否删除消息
		false,      //true 表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		nil)
	r.Err(err, "PublishSub-->交换机创建失败\n")
	//发送消息
	r.channel.Publish(
		r.Exchange,
		"",
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

//ConsumeSub ..订阅模式下消费者
func (r *RabbitMQ) ConsumeSub() {
	//1.尝试创建交换机
	err := r.channel.ExchangeDeclare(r.Exchange, "fanout", true, false, false, false, nil)
	r.Err(err, "ConsumeSub-->交换机创建失败\n")
	//2.创建队列
	q, err := r.channel.QueueDeclare(
		"",    //随机产生队列名称
		false, //控制消息是否持久化，重启服务器数据是否保留
		false, //是否自动删除，当最后一个消费者从队列离开，是否删除消息
		true,  //排他性，false时紧自己可见，其他用户不能访问
		false, //是否阻塞，
		nil)   //额外的属性
	if err != nil {
		fmt.Println(err)
	}
	err = r.channel.QueueBind(
		q.Name,     //上一步创建的队列名称
		"",         //在订阅模式下，key必须为空
		r.Exchange, //要绑定的交换机
		false,
		nil)
	//接收消息
	mesgs, err := r.channel.Consume(
		q.Name,
		"",    //区分多个消费者
		true,  //是否自动应答，
		false, //排他性
		false, //为true，不能将同一个connection中发送的消息传递给这个connection中的消费者
		false, //是否设置为阻塞
		nil)
	if err != nil {
		fmt.Println(err)
	}
	go func() {
		for d := range mesgs {
			//实现我们要处理的逻辑函数
			log.Printf("Received a message :%s", d.Body)
		}
	}()
	log.Printf("[*] waiting for messages, to exit press ctrl+c!!!")
	select {}
}

//NewRabbitMQRouting ..路由模式实例
func NewRabbitMQRouting(exchangeName string, routingkey string) *RabbitMQ {
	rbmq := NewRabbitMQ("", exchangeName, routingkey)
	return rbmq
}

//PublishRouting ...路由模式生产者
func (r *RabbitMQ) PublishRouting(message string) {
	//创建交换机(不存在则创建)
	err := r.channel.ExchangeDeclare(
		r.Exchange, //交换机名称
		"direct",   //直接类型
		true,       //控制消息是否持久化，重启服务器数据是否保留
		false,      //是否自动删除，当最后一个消费者从队列离开，是否删除消息
		false,      //true 表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		nil)
	r.Err(err, "PublishRouting-->交换机创建失败\n")
	//发送消息
	r.channel.Publish(
		r.Exchange,
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

//ConsumeRouting ..路由模式消费者
func (r *RabbitMQ) ConsumeRouting() {
	err := r.channel.ExchangeDeclare(r.Exchange, "direct", true, false, false, false, nil)
	r.Err(err, "ConsumeRouting-->交换机创建失败\n")
	q, err := r.channel.QueueDeclare("", false, false, true, false, nil)
	if err != nil {
		fmt.Println(err)
	}
	err = r.channel.QueueBind(
		q.Name,     //上一步创建的队列名称
		r.Key,      //绑定key
		r.Exchange, //要绑定的交换机
		false,
		nil)
	//接收消息
	mesgs, err := r.channel.Consume(q.Name, "", true, false, false, false, nil)
	if err != nil {
		fmt.Println(err)
	}
	go func() {
		for d := range mesgs {
			//实现我们要处理的逻辑函数
			log.Printf("Received a message :%s", d.Body)
		}
	}()
	log.Printf("[*] waiting for messages, to exit press ctrl+c!!!")
	select {}
}
